package cn._51doit.flink.day01;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 *
 * 使用MapFunction和和RichMapFunction
 *
 * RichMapFunction中的方法执行顺序（即Task的生命周期方法）
 *
 * open(初始化时执行一次) -> map（每来一条数据执行一次） -> close(停止job时会执行一次)
 *
 */
public class MapDemo {

    public static void main(String[] args) throws Exception {

        //在本地执行，执行环境的默认并行度为：当前机器cpu的逻辑核数据（cpu线程数）
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port", 8081);
        //StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        //env.setParallelism(8);

        //活动当前执行环境的默认并行度
        int parallelism = env.getParallelism();


        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        //map是多并行的算子，如果没有任务指定该算子的并行度，默认与执行环境的并行度保持一致
        SingleOutputStreamOperator<String> upperStream = lines.map(new MyMapFunction());


        //调用sink
        DataStreamSink<String> streamSink = upperStream.addSink(new RichSinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {
                //sink每来一条数据，会调用一次invoke方法
                //获取当前subtask的index，然后将index + 1
                //getRuntimeContext可以获取当前正在运行的subtask的很多信息
                int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
                System.out.println(indexOfThisSubtask + " > " + value);
            }
        });


        int parallelism3 = streamSink.getTransformation().getParallelism();
        System.out.println("Sink算子所在Task的并行度: " + parallelism3);

        env.execute();

    }


    private static class MyMapFunction extends RichMapFunction<String, String> {

        @Override
        public void open(Configuration parameters) throws Exception {
            //初始化数据库的连接
            System.out.println("!!!!! open method invoked in subtask : " + getRuntimeContext().getIndexOfThisSubtask());
        }

        @Override
        public String map(String s) throws Exception {
            //反复使用实现创建好的连接关联数据
            System.out.println("@@@@@ map method invoked in subtask : " + getRuntimeContext().getIndexOfThisSubtask());
            return s.toUpperCase();
        }

        @Override
        public void close() throws Exception {
            //关闭数据库连接
            System.out.println("### close method invoked in subtask : " + getRuntimeContext().getIndexOfThisSubtask());
        }
    }


}
